package com.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.*;
import java.net.URI;

public class Demo1 {

    private Configuration conf;
    private FileSystem fs;

    @Before
    public void init() throws IOException {
        conf = new Configuration(true);
        fs = FileSystem.get(URI.create("hdfs://node01:9000"), conf);
    }

    // public void 是固定写法
    @Test
    public void test() {
        System.out.println(fs);
    }

    @Test
    public void mkdir() throws IOException {
        Path path = new Path("/data/test");
        boolean flag = fs.mkdirs(path);
        System.out.println(flag);

    }

    @Test
    public void delete() throws IOException {
        Path path = new Path("/udf");
        boolean flag = fs.delete(path, true);
        System.out.println(flag);
    }

    // 上传文件
    @Test
    public void upload() throws IOException {
        // 使用相对路径，确定当前的路径
        // System.out.println(new File(".").getAbsolutePath());
        // 本地文件，上传到HDFS
        BufferedInputStream inputStream = new BufferedInputStream(new FileInputStream("data/demo.txt"));

        // 上传到HDFS的目录
        Path path = new Path("/data/test/demo.txt");
        FSDataOutputStream outputStream = fs.create(path);

        IOUtils.copyBytes(inputStream, outputStream, conf);

    }


    // 读取文件block数据
    @Test
    public void blocks() throws Exception {
        Path file = new Path("/bigdata/hadoop-2.6.5.tar.gz");
//        Path file = new Path("/data/test/demo.txt");
        // 获得文件状态对象
        FileStatus fileStatus = fs.getFileStatus(file);
        // block 大小, 128M
        long blockSize = fileStatus.getBlockSize();
        // 文件大小
        long len = fileStatus.getLen();
        // 文件所属
        String owner = fileStatus.getOwner();

        System.out.println(blockSize / 1024 / 1024);
        System.out.println(len / 1024 / 1024);
        System.out.println(owner);


        System.out.println("--------------");

        // 根据偏移量能够获取指定location信息，
        // len 是文件的总大小，也就是获取所有block的位置信息

        /*
        0,134217728,node02,node04
        134217728,49377148,node04,node03
        分布式计算：写好了wordcount代码，打jar包，使用命令执行
        在node01上执行的MR相关命令
        1. 把block1数据从 node04 下载到 node01， 并且计算
        2. 把block2数据从 node03 下载到 node01， 并且计算

        上面的方案存在的问题：
        文件下载回单机，没有发挥出分布式文件系统的优势

        在打jar包的时候，可以带依赖，也可以不带，不带依赖jar包非常小

        在node01上把jar上传到HDFS， 由 node04 从HDFS下载jar，把计算作用在本地的block1数据上
        由 node03 从HDFS下载jar，把计算作用在本地的block2数据上
        这样的话网络的IO就只有jar包的上传和下载

        数据不动，移动的是计算（jar包）

        HDFS 要支持计算向数据移动，那么HDFS 就要提供一个功能，
        可以根据指定的文件字节范围，返回部分文件的数据存放的节点

        block 默认大小是128M
         */
        BlockLocation[] blockLocations = fs.getFileBlockLocations(file, 0, 5);
        for (BlockLocation location : blockLocations) {
            System.out.println(location);
        }


        // 面向文件打开输入流，从文件起始位置读取
        FSDataInputStream inputStream = fs.open(file);

//        inputStream.seek(5);

        byte[] bys = new byte[1024];

        inputStream.read(bys);

        String str = new String(bys);
        System.out.println(str);


//        System.out.println(new String(bys));

        // 设置读取的其实位置，单位是字节数
    }


    @After
    public void close() throws IOException {
        fs.close();
    }


}
